package zheezes.eisp.collect.model;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Observable;
import java.util.Observer;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.springframework.context.support.AbstractApplicationContext;

import zheezes.eisp.data.dao.DataAccessor;
import zheezes.eisp.data.orm.CollectTask;
import zheezes.eisp.util.SpringUtil;
import zheezes.socket.PostOffice;
import zheezes.util.Log4jUtil;
import zheezes.util.ProcLock;
import zheezes.util.Util;


public abstract class CollectManager extends Observable implements Observer, Runnable {
	private static Logger logger = Logger.getLogger(CollectManager.class);
	
	private Thread cth;

	protected DataAccessor dataAccessor;
	protected PostOffice postOffice;

	private List<CollectTask> tasklist = new ArrayList<CollectTask>();
	@SuppressWarnings("rawtypes")
	private Map<CollectTaskHandler, Future> tcl = new ConcurrentHashMap<CollectTaskHandler, Future>();

	protected volatile boolean terminated = false;
	protected int interrupt = 1000;
	private int thcnt = 4;
	
	private ExecutorService es;
	
	private static final ProcLock lock = new ProcLock(new File(System.getProperty("java.io.tmpdir"), "eisp_collect.lock"));
	
	static {
		logger.debug("Loading");
	}

	public static void main(String[] args) {
		PropertyConfigurator.configure("log4j.properties");
		Log4jUtil.setLogFile("collect.log");
		final Thread mth = Thread.currentThread();
		AbstractApplicationContext context = SpringUtil.getApplicationContext();
		final CollectManager manager = (CollectManager) context.getBean("collectManager");
		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				logger.info("Process shutdown");
				manager.stop();
				try {
					mth.join(5000);
				} catch (InterruptedException e) {
					logger.error(e.getMessage(), e);
				}
			}
		});
		manager.run();
		context.destroy();
	}

	public abstract CollectTaskHandler select(CollectTask task);
	
	public void init() {
		es = Executors.newFixedThreadPool(thcnt);
		lock.addObserver(this);
	}

	@SuppressWarnings("rawtypes")
	public void run() {
		cth = Thread.currentThread();
		if(lock.lock() == 0) {
			while (!terminated) {
				if(postOffice.activeCount() > 0) {
					tasklist = dataAccessor.queryAllQueueCollectTasks();
					if (tasklist == null || tasklist.size() == 0) {
						Util.msleep(interrupt);
						continue;
					}
					logger.debug(String.format("total %d collect tasks", tasklist.size()));
					setChanged();
					notifyObservers(tasklist.size());
					for (CollectTask task : tasklist) {
						logger.debug(task.toString());
						CollectTaskHandler handler = select(task);
						if (handler != null) {
							handler.setDataAccessor(dataAccessor);
							handler.setPostOffice(postOffice);
							handler.setTask(task);
							logger.debug(handler.toString());
							try {
								tcl.put(handler, es.submit(handler));
							} catch (RejectedExecutionException e) { // shutdown
								break;
							}
						}
					}
					tasklist.clear(); // optimize for memory release
					while (tcl.size() > 0) {
						boolean avai = postOffice.activeCount() > 0 ? true : false; // if service is unavailable
						for (Entry<CollectTaskHandler, Future> entry : tcl.entrySet()) {
							CollectTaskHandler handler = entry.getKey();
							Future future = entry.getValue();
							if (future.isDone() || future.isCancelled()) {
								tcl.remove(handler);
							} else if(!avai) {
								handler.info();
								handler.terminate();
								future.cancel(true); // thread interrupt
							}
						}
						Util.msleep(interrupt);
					}
				} else {
					logger.info(String.format("Network service is unavailable, so collect manager will sleep"));
					Util.msleep(Integer.MAX_VALUE);
				}
			}
			logger.info("collect exit");
			lock.free();
		} else {
			setChanged();
			notifyObservers();
		}
	}

	public void destroy() {
		stop();
	}
	
	@SuppressWarnings("rawtypes")
	public void terminate() {
		terminated = true;
		cth.interrupt();
		es.shutdownNow();
		while (tcl.size() > 0) {
			logger.debug(String.format("there is %d task running", tcl.size()));
			for (Entry<CollectTaskHandler, Future> entry : tcl.entrySet()) {
				CollectTaskHandler handler = entry.getKey();
				Future future = entry.getValue();
				if (future.isDone() || future.isCancelled()) {
					tcl.remove(handler);
				} else {
					handler.info();
					handler.terminate();
					future.cancel(true); // thread interrupt
				}
			}
			Util.msleep(interrupt);
		}
	}

	public void stop() {
		terminate();
		// lock.free();
	}
	
	public void update(Observable o, Object arg) {
		if(o == lock) {
			// stop model
			terminate();
			// stop view
			setChanged();
			notifyObservers();
		} else if(o == postOffice) {
			if(cth != null) {
				cth.interrupt();
			}
		}
	}

	public int runTaskCount() {
		return tcl.size();
	}

	public void setDataAccessor(DataAccessor dataAccessor) {
		this.dataAccessor = dataAccessor;
	}

	public void setPostOffice(PostOffice postOffice) {
		this.postOffice = postOffice;
		postOffice.addObserver(this);
	}

	public void setThcnt(int thcnt) {
		this.thcnt = thcnt;
	}
}
